package connector

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"
	"sync"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

// MySQLSourceConnector represents a MySQL source connector
type MySQLSourceConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// db is the database connection
	db *sql.DB
	// rows is the result rows
	rows *sql.Rows
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// query is the SQL query
	query string
	// params are the query parameters
	params []interface{}
	// columns are the column names
	columns []string
	// position is the current position in the result set
	position int
	// batchSize is the batch size
	batchSize int
	// keyColumn is the key column name
	keyColumn string
	// mockMode indicates if the connector is in mock mode
	mockMode bool
	// mockData is the mock data
	mockData []map[string]interface{}
	// mockPosition is the current position in the mock data
	mockPosition int
}

// NewMySQLSourceConnector creates a new MySQL source connector
func NewMySQLSourceConnector() SourceConnector {
	return &MySQLSourceConnector{
		metrics:      make(map[string]interface{}),
		batchSize:    10,
		mockPosition: 0,
	}
}

// Initialize initializes the connector
func (c *MySQLSourceConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Check if mock mode is enabled
	if mockMode, ok := config["mock"].(bool); ok && mockMode {
		c.mockMode = true

		// Get mock data
		if mockDataStr, ok := config["mock_data"].(string); ok {
			var mockData []map[string]interface{}
			err := json.Unmarshal([]byte(mockDataStr), &mockData)
			if err != nil {
				return fmt.Errorf("failed to parse mock data: %w", err)
			}
			c.mockData = mockData
		} else if mockData, ok := config["mock_data"].([]map[string]interface{}); ok {
			c.mockData = mockData
		} else {
			// Create default mock data
			c.mockData = []map[string]interface{}{
				{"id": 1, "name": "Mock User 1", "email": "user1@example.com"},
				{"id": 2, "name": "Mock User 2", "email": "user2@example.com"},
				{"id": 3, "name": "Mock User 3", "email": "user3@example.com"},
			}
		}
	}

	// Connection string is not needed for initialization
	// It will be used when starting the connector

	// Get query
	if query, ok := config["query"].(string); ok {
		c.query = query
	} else if table, ok := config["table"].(string); ok {
		c.query = fmt.Sprintf("SELECT * FROM %s", table)
	} else {
		return fmt.Errorf("either query or table is required")
	}

	// Get batch size
	if batchSize, ok := config["batch_size"].(int); ok {
		c.batchSize = batchSize
	} else if batchSizeFloat, ok := config["batch_size"].(float64); ok {
		c.batchSize = int(batchSizeFloat)
	}

	// Get key column
	if keyColumn, ok := config["key_column"].(string); ok {
		c.keyColumn = keyColumn
	}

	// Get query parameters
	if params, ok := config["params"].([]interface{}); ok {
		c.params = params
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["records_read"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MySQLSourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// If in mock mode, just set started flag and return
	if c.mockMode {
		c.started = true
		c.mockPosition = 0
		return nil
	}

	// Get connection string
	var dsn string
	if connStr, ok := c.config["connection_string"].(string); ok {
		dsn = connStr
	} else {
		// Build connection string from individual parameters
		host := getStringConfig(c.config, "host", "localhost")
		port := getIntConfig(c.config, "port", 3306)
		user := getStringConfig(c.config, "user", "root")
		password := getStringConfig(c.config, "password", "")
		database := getStringConfig(c.config, "database", "")
		params := getStringConfig(c.config, "params", "parseTime=true")

		dsn = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?%s", user, password, host, port, database, params)
	}

	// Open database connection
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return fmt.Errorf("failed to open database connection: %w", err)
	}
	c.db = db

	// Set connection pool parameters
	maxOpenConns := getIntConfig(c.config, "max_open_conns", 10)
	maxIdleConns := getIntConfig(c.config, "max_idle_conns", 5)
	connMaxLifetime := getIntConfig(c.config, "conn_max_lifetime", 3600)

	db.SetMaxOpenConns(maxOpenConns)
	db.SetMaxIdleConns(maxIdleConns)
	db.SetConnMaxLifetime(time.Duration(connMaxLifetime) * time.Second)

	// Test connection
	if err := db.PingContext(ctx); err != nil {
		db.Close()
		return fmt.Errorf("failed to ping database: %w", err)
	}

	// Execute query
	rows, err := db.QueryContext(ctx, c.query, c.params...)
	if err != nil {
		db.Close()
		return fmt.Errorf("failed to execute query: %w", err)
	}
	c.rows = rows

	// Get column names
	columns, err := rows.Columns()
	if err != nil {
		rows.Close()
		db.Close()
		return fmt.Errorf("failed to get column names: %w", err)
	}
	c.columns = columns

	// Count total rows (if possible)
	var count int
	countQuery := fmt.Sprintf("SELECT COUNT(*) FROM (%s) AS count_query", c.query)
	err = db.QueryRowContext(ctx, countQuery, c.params...).Scan(&count)
	if err == nil {
		c.metrics["records_total"] = count
	}

	c.started = true
	c.position = 0

	return nil
}

// Stop stops the connector
func (c *MySQLSourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// If in mock mode, just reset started flag and return
	if c.mockMode {
		c.started = false
		return nil
	}

	// Close rows
	if c.rows != nil {
		c.rows.Close()
		c.rows = nil
	}

	// Close database connection
	if c.db != nil {
		c.db.Close()
		c.db = nil
	}

	c.started = false

	return nil
}

// Read reads records from the source
func (c *MySQLSourceConnector) Read(ctx context.Context) (*RecordBatch, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return nil, fmt.Errorf("connector not started")
	}

	// If in mock mode, return mock data
	if c.mockMode {
		return c.readMockData(ctx)
	}

	// Check if rows is valid
	if c.rows == nil {
		return nil, fmt.Errorf("rows not initialized")
	}

	// Read records
	records := make([]*Record, 0, c.batchSize)
	for i := 0; i < c.batchSize; i++ {
		// Check if there are more rows
		if !c.rows.Next() {
			break
		}

		// Read row
		record, err := c.readRow(ctx)
		if err != nil {
			return nil, fmt.Errorf("failed to read row: %w", err)
		}
		records = append(records, record)
		c.position++
	}

	// Check if no records were read
	if len(records) == 0 {
		return nil, nil
	}

	// Update metrics
	c.metrics["records_read"] = c.position
	c.metrics["last_read_time"] = time.Now().Unix()

	return NewRecordBatch(records), nil
}

// readMockData reads mock data
func (c *MySQLSourceConnector) readMockData(ctx context.Context) (*RecordBatch, error) {
	// Check if all mock data has been read
	if c.mockPosition >= len(c.mockData) {
		return nil, nil
	}

	// Create record batch
	batch := &RecordBatch{
		Records: make([]*Record, 0, c.batchSize),
	}

	// Read records
	count := 0
	for c.mockPosition < len(c.mockData) && count < c.batchSize {
		// Get mock data
		data := c.mockData[c.mockPosition]
		c.mockPosition++

		// Create record
		record := &Record{
			Metadata: make(map[string]interface{}),
		}

		// Set key if this is the key column
		if keyVal, ok := data[c.keyColumn]; ok {
			record.Key = []byte(fmt.Sprintf("%v", keyVal))
		}

		// Convert map to JSON
		value, err := json.Marshal(data)
		if err != nil {
			return nil, fmt.Errorf("failed to marshal record: %w", err)
		}
		record.Value = value

		// Add record to batch
		batch.Records = append(batch.Records, record)
		count++
	}

	// Update metrics
	c.metrics["records_read"] = c.metrics["records_read"].(int) + len(batch.Records)
	c.metrics["last_read_time"] = time.Now().Unix()

	return batch, nil
}

// readRow reads a row from the result set
func (c *MySQLSourceConnector) readRow(ctx context.Context) (*Record, error) {
	// Create a slice of interface{} to hold the values
	values := make([]interface{}, len(c.columns))
	valuePtrs := make([]interface{}, len(c.columns))
	for i := range values {
		valuePtrs[i] = &values[i]
	}

	// Scan the row into the slice
	if err := c.rows.Scan(valuePtrs...); err != nil {
		return nil, fmt.Errorf("failed to scan row: %w", err)
	}

	// Create a map for the record
	data := make(map[string]interface{})
	for i, col := range c.columns {
		val := values[i]

		// Handle nil values
		if val == nil {
			data[col] = nil
			continue
		}

		// Handle different types
		switch v := val.(type) {
		case []byte:
			// Try to convert to string
			data[col] = string(v)
		default:
			data[col] = v
		}
	}

	// Convert to JSON
	jsonBytes, err := json.Marshal(data)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal record: %w", err)
	}

	// Create record
	record := NewRecord(nil, jsonBytes)
	record.Timestamp = time.Now()

	// Set key if available
	if c.keyColumn != "" {
		if keyValue, ok := data[c.keyColumn]; ok && keyValue != nil {
			record.Key = []byte(fmt.Sprintf("%v", keyValue))
		}
	}

	return record, nil
}

// Commit commits the offset
func (c *MySQLSourceConnector) Commit(ctx context.Context, offset interface{}) error {
	// MySQL source does not support commit
	return nil
}

// GetMetrics gets connector metrics
func (c *MySQLSourceConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// MySQLSinkConnector represents a MySQL sink connector
type MySQLSinkConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// db is the database connection
	db *sql.DB
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// table is the table name
	table string
	// columns are the column names
	columns []string
	// insertStmt is the prepared insert statement
	insertStmt *sql.Stmt
	// updateStmt is the prepared update statement
	updateStmt *sql.Stmt
	// batchSize is the batch size
	batchSize int
	// buffer is the record buffer
	buffer []*Record
	// keyColumn is the key column name
	keyColumn string
	// insertMode is the insert mode (insert, update, upsert)
	insertMode string
	// mockMode indicates if the connector is in mock mode
	mockMode bool
	// mockRecords is the mock records
	mockRecords []*Record
}

// NewMySQLSinkConnector creates a new MySQL sink connector
func NewMySQLSinkConnector() SinkConnector {
	return &MySQLSinkConnector{
		metrics:    make(map[string]interface{}),
		batchSize:  10,
		buffer:     make([]*Record, 0),
		insertMode: "insert",
	}
}

// Initialize initializes the connector
func (c *MySQLSinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Check if mock mode is enabled
	if mockMode, ok := config["mock"].(bool); ok && mockMode {
		c.mockMode = true
		c.mockRecords = make([]*Record, 0)
	}

	// Get table name
	if table, ok := config["table"].(string); ok {
		c.table = table
	} else {
		return fmt.Errorf("table is required")
	}

	// Get columns
	if columns, ok := config["columns"].([]string); ok {
		c.columns = columns
	} else if columnsStr, ok := config["columns"].(string); ok {
		c.columns = strings.Split(columnsStr, ",")
		for i := range c.columns {
			c.columns[i] = strings.TrimSpace(c.columns[i])
		}
	}

	// Get batch size
	if batchSize, ok := config["batch_size"].(int); ok {
		c.batchSize = batchSize
	} else if batchSizeFloat, ok := config["batch_size"].(float64); ok {
		c.batchSize = int(batchSizeFloat)
	}

	// Get key column
	if keyColumn, ok := config["key_column"].(string); ok {
		c.keyColumn = keyColumn
	}

	// Get insert mode
	if insertMode, ok := config["insert_mode"].(string); ok {
		c.insertMode = insertMode
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *MySQLSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// If in mock mode, just set started flag and return
	if c.mockMode {
		c.started = true
		return nil
	}

	// Get connection string
	var dsn string
	if connStr, ok := c.config["connection_string"].(string); ok {
		dsn = connStr
	} else {
		// Build connection string from individual parameters
		host := getStringConfig(c.config, "host", "localhost")
		port := getIntConfig(c.config, "port", 3306)
		user := getStringConfig(c.config, "user", "root")
		password := getStringConfig(c.config, "password", "")
		database := getStringConfig(c.config, "database", "")
		params := getStringConfig(c.config, "params", "parseTime=true")

		dsn = fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?%s", user, password, host, port, database, params)
	}

	// Open database connection
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return fmt.Errorf("failed to open database connection: %w", err)
	}
	c.db = db

	// Set connection pool parameters
	maxOpenConns := getIntConfig(c.config, "max_open_conns", 10)
	maxIdleConns := getIntConfig(c.config, "max_idle_conns", 5)
	connMaxLifetime := getIntConfig(c.config, "conn_max_lifetime", 3600)

	db.SetMaxOpenConns(maxOpenConns)
	db.SetMaxIdleConns(maxIdleConns)
	db.SetConnMaxLifetime(time.Duration(connMaxLifetime) * time.Second)

	// Test connection
	if err := db.PingContext(ctx); err != nil {
		db.Close()
		return fmt.Errorf("failed to ping database: %w", err)
	}

	// Get table columns if not provided
	if len(c.columns) == 0 {
		query := fmt.Sprintf("SHOW COLUMNS FROM %s", c.table)
		rows, err := db.QueryContext(ctx, query)
		if err != nil {
			db.Close()
			return fmt.Errorf("failed to get table columns: %w", err)
		}
		defer rows.Close()

		c.columns = make([]string, 0)
		for rows.Next() {
			var field, fieldType, null, key, defaultValue, extra sql.NullString
			if err := rows.Scan(&field, &fieldType, &null, &key, &defaultValue, &extra); err != nil {
				db.Close()
				return fmt.Errorf("failed to scan column: %w", err)
			}
			c.columns = append(c.columns, field.String)
		}
	}

	// Prepare insert statement
	if c.insertMode == "insert" || c.insertMode == "upsert" {
		placeholders := make([]string, len(c.columns))
		for i := range placeholders {
			placeholders[i] = "?"
		}
		query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)",
			c.table,
			strings.Join(c.columns, ", "),
			strings.Join(placeholders, ", "))

		stmt, err := db.PrepareContext(ctx, query)
		if err != nil {
			db.Close()
			return fmt.Errorf("failed to prepare insert statement: %w", err)
		}
		c.insertStmt = stmt
	}

	// Prepare update statement if needed
	if (c.insertMode == "update" || c.insertMode == "upsert") && c.keyColumn != "" {
		setClauses := make([]string, 0, len(c.columns))
		for _, col := range c.columns {
			if col != c.keyColumn {
				setClauses = append(setClauses, fmt.Sprintf("%s = ?", col))
			}
		}
		query := fmt.Sprintf("UPDATE %s SET %s WHERE %s = ?",
			c.table,
			strings.Join(setClauses, ", "),
			c.keyColumn)

		stmt, err := db.PrepareContext(ctx, query)
		if err != nil {
			if c.insertStmt != nil {
				c.insertStmt.Close()
			}
			db.Close()
			return fmt.Errorf("failed to prepare update statement: %w", err)
		}
		c.updateStmt = stmt
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *MySQLSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// If in mock mode, just reset started flag and return
	if c.mockMode {
		c.started = false
		return nil
	}

	// Flush any buffered records
	if len(c.buffer) > 0 {
		if err := c.flushBuffer(ctx); err != nil {
			return fmt.Errorf("failed to flush buffer: %w", err)
		}
	}

	// Close prepared statements
	if c.insertStmt != nil {
		c.insertStmt.Close()
		c.insertStmt = nil
	}
	if c.updateStmt != nil {
		c.updateStmt.Close()
		c.updateStmt = nil
	}

	// Close database connection
	if c.db != nil {
		c.db.Close()
		c.db = nil
	}

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *MySQLSinkConnector) Write(ctx context.Context, batch *RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// If in mock mode, just store records and return
	if c.mockMode {
		c.mockRecords = append(c.mockRecords, batch.Records...)
		c.metrics["records_total"] = c.metrics["records_total"].(int) + len(batch.Records)
		c.metrics["last_write_time"] = time.Now().Unix()
		return nil
	}

	// Add records to buffer
	c.buffer = append(c.buffer, batch.Records...)

	// Flush buffer if it's full
	if len(c.buffer) >= c.batchSize {
		if err := c.flushBuffer(ctx); err != nil {
			return fmt.Errorf("failed to flush buffer: %w", err)
		}
	}

	return nil
}

// flushBuffer flushes the buffer to the database
func (c *MySQLSinkConnector) flushBuffer(ctx context.Context) error {
	// Check if buffer is empty
	if len(c.buffer) == 0 {
		return nil
	}

	// Start a transaction
	tx, err := c.db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer func() {
		if err != nil {
			tx.Rollback()
		}
	}()

	// Process each record
	for _, record := range c.buffer {
		// Parse record value
		var data map[string]interface{}
		if err := json.Unmarshal(record.Value, &data); err != nil {
			return fmt.Errorf("failed to parse record value: %w", err)
		}

		// Extract values for columns
		values := make([]interface{}, len(c.columns))
		for i, col := range c.columns {
			if val, ok := data[col]; ok {
				values[i] = val
			} else {
				values[i] = nil
			}
		}

		// Handle different insert modes
		switch c.insertMode {
		case "insert":
			// Execute insert statement
			if _, err := tx.StmtContext(ctx, c.insertStmt).ExecContext(ctx, values...); err != nil {
				return fmt.Errorf("failed to execute insert statement: %w", err)
			}
		case "update":
			if c.keyColumn == "" {
				return fmt.Errorf("key_column is required for update mode")
			}

			// Find key column index
			keyIndex := -1
			for i, col := range c.columns {
				if col == c.keyColumn {
					keyIndex = i
					break
				}
			}
			if keyIndex == -1 {
				return fmt.Errorf("key column %s not found in columns", c.keyColumn)
			}

			// Extract key value
			keyValue := values[keyIndex]
			if keyValue == nil {
				return fmt.Errorf("key value is nil")
			}

			// Extract non-key values
			updateValues := make([]interface{}, 0, len(values))
			for i, val := range values {
				if i != keyIndex {
					updateValues = append(updateValues, val)
				}
			}
			// Add key value at the end for WHERE clause
			updateValues = append(updateValues, keyValue)

			// Execute update statement
			if _, err := tx.StmtContext(ctx, c.updateStmt).ExecContext(ctx, updateValues...); err != nil {
				return fmt.Errorf("failed to execute update statement: %w", err)
			}
		case "upsert":
			if c.keyColumn == "" {
				return fmt.Errorf("key_column is required for upsert mode")
			}

			// Find key column index
			keyIndex := -1
			for i, col := range c.columns {
				if col == c.keyColumn {
					keyIndex = i
					break
				}
			}
			if keyIndex == -1 {
				return fmt.Errorf("key column %s not found in columns", c.keyColumn)
			}

			// Extract key value
			keyValue := values[keyIndex]
			if keyValue == nil {
				// If key is nil, just insert
				if _, err := tx.StmtContext(ctx, c.insertStmt).ExecContext(ctx, values...); err != nil {
					return fmt.Errorf("failed to execute insert statement: %w", err)
				}
				continue
			}

			// Check if record exists
			var exists bool
			query := fmt.Sprintf("SELECT 1 FROM %s WHERE %s = ? LIMIT 1", c.table, c.keyColumn)
			err := tx.QueryRowContext(ctx, query, keyValue).Scan(&exists)
			if err != nil && err != sql.ErrNoRows {
				return fmt.Errorf("failed to check if record exists: %w", err)
			}

			if err == sql.ErrNoRows {
				// Record doesn't exist, insert
				if _, err := tx.StmtContext(ctx, c.insertStmt).ExecContext(ctx, values...); err != nil {
					return fmt.Errorf("failed to execute insert statement: %w", err)
				}
			} else {
				// Record exists, update
				// Extract non-key values
				updateValues := make([]interface{}, 0, len(values))
				for i, val := range values {
					if i != keyIndex {
						updateValues = append(updateValues, val)
					}
				}
				// Add key value at the end for WHERE clause
				updateValues = append(updateValues, keyValue)

				// Execute update statement
				if _, err := tx.StmtContext(ctx, c.updateStmt).ExecContext(ctx, updateValues...); err != nil {
					return fmt.Errorf("failed to execute update statement: %w", err)
				}
			}
		default:
			return fmt.Errorf("unsupported insert mode: %s", c.insertMode)
		}
	}

	// Commit transaction
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("failed to commit transaction: %w", err)
	}

	// Update metrics
	c.metrics["records_total"] = c.metrics["records_total"].(int) + len(c.buffer)
	c.metrics["last_write_time"] = time.Now().Unix()

	// Clear buffer
	c.buffer = c.buffer[:0]

	return nil
}

// Flush flushes any buffered records
func (c *MySQLSinkConnector) Flush(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// If in mock mode, just return
	if c.mockMode {
		return nil
	}

	// Flush buffer
	if len(c.buffer) > 0 {
		if err := c.flushBuffer(ctx); err != nil {
			return fmt.Errorf("failed to flush buffer: %w", err)
		}
	}

	return nil
}

// GetMetrics gets connector metrics
func (c *MySQLSinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// Helper functions

// getStringConfig gets a string configuration value with a default
func getStringConfig(config map[string]interface{}, key, defaultValue string) string {
	if value, ok := config[key].(string); ok {
		return value
	}
	return defaultValue
}

// getIntConfig gets an integer configuration value with a default
func getIntConfig(config map[string]interface{}, key string, defaultValue int) int {
	if value, ok := config[key].(int); ok {
		return value
	}
	return defaultValue
}

// getBoolConfig gets a boolean configuration value with a default
func getBoolConfig(config map[string]interface{}, key string, defaultValue bool) bool {
	if value, ok := config[key].(bool); ok {
		return value
	}
	return defaultValue
}
